package com.shujia.flink.core

import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import scala.collection.mutable

object Demo9FraudTimer {
  def main(args: Array[String]): Unit = {
    /**
     * 对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
     * 001,20.1
     * 001,30
     * 001,0.07
     * 001,100
     * 001,800
     * 001,20
     * 001,0.1
     * 001,700
     * 001,20
     * 001,0.1
     * 002,700
     * 001,800
     */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    //解析数据
    val kvDS: DataStream[(String, Double)] = linesDS.map(line => {
      val split: Array[String] = line.split(",")
      val id: String = split(0)
      val price: Double = split(1).toDouble
      (id, price)
    })
    //安装账户分组
    val keyByDS: KeyedStream[(String, Double), String] = kvDS.keyBy(_._1)

    val foaudDS: DataStream[Fraud] = keyByDS
      .process(new KeyedProcessFunction[String, (String, Double), Fraud] {

        //通过task中的数据共享同一个变量
        //var flag = false
        //var minPrice: Double = _

        //需要为每一个key保存一个flag和minPrice
        val flagMap = new mutable.HashMap[String, Boolean]()
        val minPriceMap = new mutable.HashMap[String, Double]()

        /**
         * 对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
         *
         * processElement: 每一条数据执行一次
         *
         * @param value ： 一行数据
         * @param ctx   ： 上下文对象
         * @param out   ： 用于将数据发送到下游
         */
        override def processElement(value: (String, Double),
                                    ctx: KeyedProcessFunction[String, (String, Double), Fraud]#Context,
                                    out: Collector[Fraud]): Unit = {
          val (id: String, price: Double) = value

          //从flagMap获取flag值
          val flag: Boolean = flagMap.getOrElse(id, false)

          if (flag) {
            //当出现大于500的金额之后发出告警信息
            if (price > 500) {
              //获取当前时间
              val time: Long = System.currentTimeMillis()
              //取出小的金额
              val minPrice: Double = minPriceMap.getOrElse(id, 0.0)
              //发出告警信息
              out.collect(Fraud(id, time, minPrice, price))
            }
            //不管有没有发出告警信息都需要重置flag
            flagMap.put(id, false)
          }
          //当金额小于1时，将flag改成true，同时保存金额
          if (price < 1) {
            flagMap.put(id, true)
            minPriceMap.put(id, price)

            //获取当前的时间
            val time: Long = ctx.timerService().currentProcessingTime()
            //创建定时器,再5秒之后会触发onTimer的执行
            ctx.timerService().registerProcessingTimeTimer(time + 5000)
          }
        }

        //当时间到了会触发执行的方法
        override def onTimer(timestamp: Long,
                             ctx: KeyedProcessFunction[String, (String, Double), Fraud]#OnTimerContext,
                             out: Collector[Fraud]): Unit = {
          //获取当前的key
          val id: String = ctx.getCurrentKey
          //将flag设置为false
          flagMap.put(id, false)
        }
      })

    foaudDS.print()

    env.execute()

  }

  case class Fraud(id: String, time: Long, minPrice: Double, maxPrice: Double)

}
